package com.meshnetics.gb.stack.mac.util;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

/**
 * Pool of thread above of queue of execution tasks. When new execution task is submitted and no threads are free it is
 * stored in thread. Each thread when it becomes free (finished exeution of task) checks queue for pending tasks and
 * takes one for execution or waits for tasks to become available. Executor provides autopmatically relaunchs threads
 * that died from uncaught exceptions or errors.
 *
 * @author AKovylin
 */
public class ThreadPoolExecutor {
    private final Log log = LogFactory.getLog(ThreadPoolExecutor.class);

    private final class Queue {
        private final Object syncObject = new Object();
        private final LinkedList queue = new LinkedList();
        private final LinkedList urgentQueue = new LinkedList();

        public void put(Runnable r) {
            synchronized (syncObject) {
                queue.addLast(r);
                syncObject.notify();
            }
        }

        public void urgent(Runnable r) {
            synchronized (syncObject) {
                urgentQueue.addLast(r);
                syncObject.notify();
            }
        }

        public Runnable pick() throws InterruptedException {
            synchronized (syncObject) {
                while (urgentQueue.size() + queue.size() == 0) {
                    syncObject.wait();
                }
                return urgentQueue.size() > 0 ? (Runnable) urgentQueue.removeFirst() : (Runnable) queue.removeFirst();
            }
        }

        public void clear() {
            synchronized (syncObject) {
                queue.clear();
                urgentQueue.clear();
            }
        }
    }

    private final class ExecutorThreadGroup extends ThreadGroup {
        public ExecutorThreadGroup(String name) {
            super(name);
        }

        public void uncaughtException(Thread t, Throwable e) {
            log.error("uncought exception in executor, relaunching", e);
            launchDeadExecutor(t);
        }
    }

    private final class Executor extends Thread {
        public Executor(ThreadGroup threadGroup, String name) {
            super(threadGroup, name);
        }

        public void run() {
            Runnable r;
            while (!isInterrupted()) {
                try {
                    r = queue.pick();
                } catch (InterruptedException e) {
                    break;
                }
                try {
                    r.run();
                } catch (RuntimeException e) {
                    log.error("execution failed", e);
                }
            }
            if (log.isDebugEnabled()) {
                log.debug("thread " + this + " finished");
            }
        }
    }

    private static int threadNameNum = 0;
    private final Queue queue = new Queue();
    private final ExecutorThreadGroup threadGroup;
    private final String groupName;
    private int threadNum = 0;
    private final List executors;
    private boolean dead = false;
    private boolean blocked = false;

    public ThreadPoolExecutor(int poolSize) {
        this(null, poolSize);
    }

    public ThreadPoolExecutor(String name, int poolSize) {
        if (poolSize < 1) {
            throw new IllegalArgumentException("poolSize must be positive number");
        }
        this.executors = new ArrayList(poolSize);
        this.groupName = name == null ? "ThreadPoolExecutor-" + threadNameNum++ : name;
        threadGroup = new ExecutorThreadGroup(groupName);
        for (int i = 0; i < poolSize; i++) {
            Executor e = new Executor(threadGroup, groupName + '-' + threadNum++);
            executors.add(e);
            e.start();
        }
    }

    public int getPoolSize() {
        synchronized (executors) {
            return executors.size();
        }
    }

    private void launchDeadExecutor(Thread t) {
        synchronized (executors) {
            if (!dead) {
                int index = executors.indexOf(t);
                if (index == -1) {
                    log.error("dead executor reproted but not found in executors list");
                } else {
                    Executor e = new Executor(threadGroup, groupName + '-' + threadNum++);
                    executors.set(index, e);
                    e.start();
                }
            }
        }
    }

    public boolean execute(Runnable r) {
        synchronized (executors) {
            if (!blocked && !dead) {
                this.queue.put(r);
                return true;
            } else {
                return false;
            }
        }
    }

    public boolean executeFast(Runnable r) {
        synchronized (executors) {
            if (!blocked && !dead) {
                this.queue.urgent(r);
                return true;
            } else {
                return false;
            }
        }
    }

    public void kill() {
        synchronized (executors) {
            if (!dead) {
                Iterator it = executors.iterator();
                while (it.hasNext()) {
                    ((Executor) it.next()).interrupt();
                }
                dead = true;
            }
        }
    }

    public void clearQueues() {
        synchronized (executors) {
            this.queue.clear();
        }
    }

    public boolean isBlocked() {
        synchronized (executors) {
            return blocked;
        }
    }

    public void setBlocked(boolean blocked) {
        synchronized (executors) {
            this.blocked = blocked;
        }
    }

    public void join() throws InterruptedException {
        synchronized (executors) {
            if (dead) {
                Iterator it = executors.iterator();
                while (it.hasNext()) {
                    ((Executor) it.next()).join();
                }
            } else {
                throw new IllegalStateException("pool of threads can be joined only after executor is killed");
            }
        }
    }
}
